-----取标准时效
insert  overwrite table jms_tmp.tmp_cainiao_sign_standard_bak
SELECT
    in_from_code
     ,max(in_from_name) as in_from_name
     ,out_to_code
     ,max(out_to_name) as out_to_name
     ,out_sign_days
from
    (
        select
            in_from_code
             ,in_from_name
             ,out_to_code
             ,out_to_name
             ,out_sign_days
        from jms_dm.dm_cainiao_route_completion_data
        where dt='{{ execution_date  | cst_ds }}'
        union all
        select
            in_from_code
             ,in_from_name
             ,out_to_code
             ,out_to_name
             ,out_sign_days
        from jms_dm.dm_cainiao_route_push_old_data
    )t group by in_from_code,out_to_code,out_sign_days
distribute by pmod(hash(rand()), 20)
;

-------------------------------------------------------------------------------------------
insert overwrite table jms_tmp.tmp_cainiao_history_standard_res_dt_bak
select
    taking_code
     ,taking_name
     ,deliver_code
     ,deliver_name
     ,start_hm
     ,end_hm
     ,out_sign_days as days
     ,'{{ execution_date  | cst_ds }}' as date_time
     ,rt
     ,all_bill --当前时段总单量
     ,min(start_hm) over(partition by taking_code,deliver_code) as hm_min
     ,max(end_hm) over(partition by taking_code,deliver_code) as hm_max
     ,first_value(out_sign_days) over(partition by taking_code,deliver_code order by out_sign_days ) as allhm_top_days
     ,'{{ execution_date  | cst_ds }}' as dt
from
    (
        select  --取靠近标准时效的逻辑
            taking_code
             ,taking_name
             ,deliver_code
             ,deliver_name
             ,tn
             ,start_hm
             ,end_hm
             ,out_sign_days
             ,rt
             ,row_number() over(partition by taking_code,deliver_code,start_hm,end_hm order by diff_abs,out_sign_days) as  abs_flag
              ,all_bill
        from
            (
                select
                    t1.taking_code
                     ,t1.taking_name
                     ,t1.deliver_code
                     ,t1.deliver_name
                     ,t1.tn
                     ,t1.start_hm
                     ,t1.end_hm
                     ,t2.out_sign_days
                     ,t1.rt
                     ,abs(cast((t1.tn-t2.out_sign_days) as int)) as diff_abs
                     ,all_bill
                from
                    (select * from  jms_tmp.tmp_actual_history_network_tn_dt where dt='{{ execution_date  | cst_ds }}' and rk=1) t1
                        inner join jms_tmp.tmp_cainiao_sign_standard_bak t2
                                   on t1.taking_code=t2.in_from_code and t1.deliver_code=t2.out_to_code

            )t3
    )t4
where abs_flag=1;
-------------------------------------------------------------------------------------------------------------

with shift_all as (
--例如1：（08:00-17:00），则补全为（0:00-07:59）\（08:00-17:00）\（17:01-23:59）
--例如2：（08:00-16:00）（16:01-22:30），则补全为 （0:00-07:59）\（08:00-16:00）（16:01-22:30）和（22:31-23:59）
    select
        taking_code,deliver_code,start_hm_2 as start_hm,end_hm_2 as end_hm
    from
        (
            select
                taking_code,deliver_code,start_hm,end_hm,start_hm_arr
                 ,case when ARRAY_CONTAINS(start_hm_arr,start_hm)!=1 and start_hm='00:00' then '00:00' 
          when ARRAY_CONTAINS(start_hm_arr,start_hm)!=1 then from_unixtime(unix_timestamp(concat('{{ execution_date  | cst_ds }} ',start_hm,':59'))+60,'HH:mm')
          else start_hm
     end as start_hm_2
    ,case when ARRAY_CONTAINS(start_hm_arr,end_hm)=1  then from_unixtime(unix_timestamp(concat('{{ execution_date  | cst_ds }} ',end_hm,':59'))-60,'HH:mm')
          else end_hm
     end as end_hm_2
            from
                (
                SELECT
                taking_code,deliver_code,start_hm
                    ,collect_set(if(flag=1,start_hm,null)) over(partition by taking_code,deliver_code) as start_hm_arr
                    ,lead(start_hm,1,'23:59') over(partition by taking_code,deliver_code order by start_hm) as end_hm
                from
                (
                SELECT
                taking_code,deliver_code,start_hm,flag,
                row_number() over(partition by taking_code,deliver_code,start_hm order by flag) as rk
                from
                (
                select taking_code,deliver_code,start_hm, 1 as flag from jms_tmp.tmp_cainiao_history_standard_res_dt_bak where dt='{{ execution_date  | cst_ds }}'
                union all
                select taking_code,deliver_code,end_hm  , 2 as flag from jms_tmp.tmp_cainiao_history_standard_res_dt_bak where dt='{{ execution_date  | cst_ds }}'
                union all
                select taking_code,deliver_code,'00:00'  , 3 as flag from jms_tmp.tmp_cainiao_history_standard_res_dt_bak where dt='{{ execution_date  | cst_ds }}'
                union all
                select taking_code,deliver_code,'23:59'  , 3 as flag from jms_tmp.tmp_cainiao_history_standard_res_dt_bak where dt='{{ execution_date  | cst_ds }}'
                )t
                )t2 where rk=1
                )t3 where start_hm!=end_hm
        )t4 where start_hm_2<=end_hm_2
)

insert overwrite table jms_tmp.tmp_cainiao_actual_network_tn_dt_bak
select
    taking_code
     ,taking_name
     ,deliver_code
     ,deliver_name
     ,start_hm
     ,end_hm
     ,days
     ,date_time
     ,rt
     ,dt
from
    (   select
            taking_code
             ,taking_name
             ,deliver_code
             ,deliver_name
             ,start_hm
             ,end_hm
             ,nvl(days,jing_tn) as days --补充时段时效为空取静态最小
             ,date_time
             ,rt
             ,dt
        from
            (
                select
                    shift.taking_code
                     ,net1.name as taking_name
                     ,shift.deliver_code
                     ,net2.name as deliver_name
                     ,shift.start_hm
                     ,shift.end_hm
                     ,nvl(t5.days,t5.allhm_top_days) as days
                     ,'{{ execution_date  | cst_ds }}' as date_time
                     ,t5.rt
                     ,'{{ execution_date  | cst_ds }}' as  dt
                     --allhm_top_days 为null时  first_value()取到的是null
                     ,first_value(allhm_top_days) over(partition by shift.taking_code,shift.deliver_code order by if(allhm_top_days is null,999,allhm_top_days) ) as jing_tn
                from shift_all shift
                         left join (select * from jms_tmp.tmp_cainiao_history_standard_res_dt_bak where dt='{{ execution_date  | cst_ds }}' )  t5
                                   on shift.taking_code=t5.taking_code and shift.deliver_code=t5.deliver_code and shift.start_hm=t5.start_hm  and shift.end_hm=t5.end_hm
                         left join jms_dim.dim_network_whole_massage net1 on net1.code=shift.taking_code
                         left join jms_dim.dim_network_whole_massage net2 on net2.code=shift.deliver_code

            )t

        union  all
        --补充历史没有的时效 但是静态路由有的时效
        select
            in_from_code as taking_code
             ,in_from_name as taking_name
             ,out_to_code as deliver_code
             ,out_to_name as deliver_name
             ,'00:00' as start_hm
             ,'23:59' as end_hm
             ,out_sign_days as days
             ,'{{ execution_date  | cst_ds }}' as date_time
             ,null as rt
             ,'{{ execution_date  | cst_ds }}' as dt
        from
            (
                SELECT
                    in_from_code
                     ,max(in_from_name) as in_from_name
                     ,out_to_code
                     ,max(out_to_name) as out_to_name
                     ,min(out_sign_days) out_sign_days
                from jms_tmp.tmp_cainiao_sign_standard_bak
                group by in_from_code,out_to_code
            )t1
                left join
            (
                select taking_code ,deliver_code
                from  jms_tmp.tmp_actual_history_network_tn_dt
                where dt='{{ execution_date  | cst_ds }}'
                group by taking_code ,deliver_code
            )t2 on t1.in_from_code=t2.taking_code and t1.out_to_code=t2.deliver_code
        where t2.taking_code is null
    )tt
    where (deliver_code  is not null and taking_code is not null and days is not null and start_hm is not null and end_hm  is not null)
    distribute by pmod(hash(rand()), 60);